package com.share.bigdata.flink.countword.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.share.bigdata.flink.countword.sources.JsonStringSourceFunction;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

public class CountWordApp {

    public static void main(String[] args) {
        try {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

            // source
//            DataStream<String> stream = env.addSource(CommonConf.getKafkaConsumerSource());
            DataStream<String> stream = env.addSource(new JsonStringSourceFunction());
            System.out.println(CountWordApp.class.getName() + "---开始接收数据");
            stream.print("kafka");

            //数据过滤
            DataStream<String> filterStream = filter(stream);

            // transform
            DataStream<Tuple2<String, Integer>> kvStream = filterStream.map(JSON::parseObject).map(new RichMapFunction<JSONObject, Tuple2<String, Integer>>() {

                @Override
                public Tuple2 map(JSONObject jsonObject) throws Exception {
                    String productcode = jsonObject.getString("productCode");
                    return new Tuple2<String, Integer>(productcode, 1);
                }
            });


            KeyedStream<Tuple2<String, Integer>, Tuple> groupStream = kvStream.keyBy(0);
            groupStream.print("groupStream");
            groupStream.timeWindow(Time.seconds(5), Time.seconds(1)).sum(1).print("jy");
//            WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> winStream = kyStream.timeWindow(Time.seconds(3000));
//            SingleOutputStreamOperator<Tuple2<String, Integer>> sumStream = winStream.sum(1);

//            WindowedStream<Tuple2<String, Integer>, Tuple, TimeWindow> sum = kyStream.timeWindow(Time.seconds(3000));
//            AllWindowedStream<Tuple2<String, Integer>, TimeWindow> fdf = kyStream.tes  sum(1).timeWindowAll(Time.seconds(3000));
//            fdf.print("sumStream");
//            sum.aggregate(AggregateS);
            // 将结果写入redis
//            sumStream.addSink(new RedisSink(CommonConf.getFlinkJedisPoolConfig(), new HashSetRedisMapper()));

            // 执行任务
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static DataStream<String> filter(DataStream<String> source) {
        return source.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                Long id = jsonObject.getLong("id");

                return id % 2 == 0;
            }
        });
    }

}
